package com.bfxy.kafka.api.quickstart;

import com.bfxy.kafka.api.constant.Const;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

public class QuickStartConsumer {

    public static void main(String[] args) {
        // 1.配置属性参数
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.218.21:9092");
        // 字符串反序列化类（可以直接用后面这串字符串）：org.apache.kafka.common.serialization.StringDeserializer
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 非常重要的属性配置：与消费者订阅组有关系
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "quickstart-group");
        // 常规属性：会话连接超时时间，默认是10000
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
        // 消费者提交offset：自动提交&手动提交，默认是自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        // 自动提交的提交周期，默认5000
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);

        // 2.创建消费者对象
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
            // 3.订阅感兴趣的主题
            consumer.subscribe(Collections.singletonList(Const.TOPIC_QUICKSTART));
            System.out.println("quickstart consumer started...");
            // 4.采用拉取消息的方式消费数据
            while (true) {
                // 设置等待多久拉取一次消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                // 拉取Const.TOPIC_QUICKSTART主题里面所有的消息
                // topic和partition是一对多的关系，一个topic可以有多个partition
                // 因为消息是在partition里存储的，所以需要遍历partition集合
                for (TopicPartition partition : records.partitions()) {
                    // 通过TopicPartition获取指定的消息集合，获取到的就是当前TopicPartition里面所有的消息
                    List<ConsumerRecord<String, String>> partitionRecordList = records.records(partition);
                    // 获取当前TopicPartition对应的主题名称
                    String topic = partition.topic();
                    // 获取当前TopicPartition下的消息条数
                    int size = partitionRecordList.size();

                    System.out.println(String.format("--- 获取topic：%s，分区位置：%s，消息总数：%s ---", topic, partition.partition(), size));

                    for (ConsumerRecord<String, String> consumerRecord : partitionRecordList) {
                        // 实际的消息内容
                        String value = consumerRecord.value();
                        // 当前获取的消息偏移量
                        long offset = consumerRecord.offset();
                        // 提交的消息偏移量：如果要提交的话，必须提交当前消息的offset+1，表示下一次从什么位置（offset）拉取消息
                        long commitOffset = offset + 1;
                        System.out.println(String.format("-- 获取实际消息value：%s，消息offset：%s，提交offset：%s ---", value, offset, commitOffset));
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
